package com.k.activemq.openwire3;

import com.k.activemq.ActiveMQUtils;
import com.k.platform.util.DateUtil;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;
import java.nio.charset.StandardCharsets;
import java.time.ZoneId;

public class PubSubUtil {
    public static ActiveMQConnectionFactory getActiveMQConnectionFactory(String url) {
        // tcp://192.168.0.169:22000
        // failover:(tcp://primary:61616,tcp://secondary:61616)?randomize=false
        ActiveMQConnectionFactory connectionFactory = ActiveMQUtils.getActiveMQConnectionFactory(url);
//        connectionFactory.setClientID(clientID);
        return connectionFactory;
    }

    public static Connection getConnection(ActiveMQConnectionFactory connectionFactory, String username, String password) throws JMSException {
        // 创建连接
        Connection connection = connectionFactory.createConnection(username, password);
        // 打开连接
        connection.start();
        return connection;
    }

    public static Session getSession(Connection connection) throws JMSException {
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        return session;
    }

    public static Destination getDestination(Session session, String destinationName, boolean pubSubDomain) throws JMSException {
        Destination destination = pubSubDomain ? session.createTopic(destinationName) : session.createQueue(destinationName);
        return destination;
    }

    public static MessageConsumer getMessageConsumer(Session session, Destination destination) throws JMSException {
        MessageConsumer consumer = session.createConsumer(destination);
        return consumer;
    }

    public static MessageProducer getMessageProducer(Session session, Destination destination) throws JMSException {
        MessageProducer producer = session.createProducer(destination);
        return producer;
    }

    public static void formatResponse(Message message) {
        try{
            if (message instanceof TextMessage) {
                TextMessage textMessage = (TextMessage) message;
                String msgStr = textMessage.getText();
                System.out.println(ActiveMQUtils.formatMsg("TextMessage", Thread.currentThread().toString(), msgStr));
            } else if (message instanceof BytesMessage) {
                BytesMessage byteMessage = (BytesMessage) message;
                String msgStr = ActiveMQUtils.byteMessageToString(byteMessage);
                System.out.println(ActiveMQUtils.formatMsg("BytesMessage", Thread.currentThread().toString(), msgStr));

            } else if (message instanceof MapMessage) {
                MapMessage mapMessage = (MapMessage) message;
                String msgStr = mapMessage.toString();
                System.out.println(ActiveMQUtils.formatMsg("MapMessage", Thread.currentThread().toString(), msgStr));
            } else if (message instanceof ObjectMessage) {
                String msgStr = "";
                System.out.println(ActiveMQUtils.formatMsg("ObjectMessage", Thread.currentThread().toString(), msgStr));
            } else if (message instanceof StreamMessage) {
                String msgStr = "";
                System.out.println(ActiveMQUtils.formatMsg("StreamMessage", Thread.currentThread().toString(), msgStr));
            } else {
                System.out.println("消息类型：Unknown");
            }
        }catch (Exception e){
            e.printStackTrace();
        }

    }

    public static void receiveAsync(MessageConsumer consumer) throws JMSException {
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                System.out.println(DateUtil.nowStr(DateUtil.PATTERN_SS, ZoneId.of("GMT+8")) + "监听到一条消息：");
                System.out.println(Thread.currentThread() + ": message:" + message);
                formatResponse(message);
                System.out.println("监听完成！");
            }
        });
    }

    public static void sendMessage(MessageProducer producer, Message message) throws JMSException {
        //发送消息
        producer.send(message);
    }

    /**
     * 创建消息
     *
     * @param session session
     * @param msg     msg
     * @param b       true:byte message
     * @return Message
     * @throws JMSException e
     */
    public static Message createMessage(Session session, String msg, boolean b) throws JMSException {
        Message message;
        if (b) {
            BytesMessage bytesMessage = session.createBytesMessage();
            byte[] byteMsg = msg.getBytes(StandardCharsets.UTF_8);
            bytesMessage.writeBytes(byteMsg);
            message = bytesMessage;
        } else {
            message = session.createTextMessage(msg);
        }
        message.setJMSDeliveryMode(DeliveryMode.PERSISTENT);
        return message;
    }
    public static Message createMessage(Session session,String msg) throws JMSException {
        return createMessage(session,msg,false);
    }

    public static void close(MessageProducer producer,MessageConsumer consumer,Session session,Connection connection){
        if(producer!=null){
            try {
                producer.close();
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
        if(consumer!=null){
            try {
                consumer.close();
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
        if(session!=null){
            try {
                session.close();
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
        if(connection!=null){
            try {
                connection.close();
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }
}
